package com.lrd.rmqcluster.rmq5;

import com.alibaba.fastjson.JSON;
import com.lrd.rmqcluster.rmq5.testconsumer.TestMessageProcessorService;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class MQMessageListener implements MessageListenerConcurrently {

    private static final Logger log = LoggerFactory.getLogger(MQMessageListener.class);

    @Autowired
    private TestMessageProcessorService testMessageProcessorService;

    @Value("${apache.rocketmq.test.topic}")
    private String testTopic;

    @Override
    public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
        try {
            for (MessageExt msg : msgs) {
                String msgBody = new String(msg.getBody(), RemotingHelper.DEFAULT_CHARSET);
                String topic = msg.getTopic();
                log.info("MQListener|consumeMessage|开始处理消息:topic:{},tag:{},msgId:{},msgKey:{},msgBody:{}", topic, msg.getTags(), msg.getMsgId(), msg.getKeys(), msgBody);
                boolean consumerSuccess;
                if (topic.equals(testTopic)) {
                    consumerSuccess = testMessageProcessorService.handleMessage(msg);
                } else {
                    //没有匹配的topic直接返回成功
                    log.info("MQListener|consumeMessage|no match topic");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
                if (!consumerSuccess) {
                    log.warn("MQListener|MQ消费失败:{},msg{}", consumerSuccess, msg);
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                }
                //return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        } catch (Exception e) {
            log.error("MQListener|consumeMessage消费端:消费MQ失败msgContext: {}", JSON.toJSONString(msgs), e);
            return ConsumeConcurrentlyStatus.RECONSUME_LATER;
        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
